package com.cctc.rds.scheduler.client.raft.rpc;

import com.cctc.rds.nrpc.api.exception.RpcIOException;
import com.cctc.rds.nrpc.api.rpc.RpcRequest;
import com.cctc.rds.nrpc.api.rpc.RpcResponse;
import com.cctc.rds.scheduler.client.raft.NettyRpcClientHolder;
import com.cctc.rds.scheduler.core.raft.TaskRpc;
import com.cctc.rds.scheduler.core.task.TaskLog;

import java.util.List;

/**
 * 任务实例日志 RPC 客户端 [请求] 实现类。
 * 主流程：存储（save）-> 取消（cancel）-> 拉取执行（pull）-> 提交执行结果（commit）。
 */
public class TaskRpcRequestImpl implements TaskRpc {
    private NettyRpcClientHolder clientHolder;

    public TaskRpcRequestImpl(NettyRpcClientHolder clientHolder) {
        this.clientHolder = clientHolder;
    }

    @Override
    public long save(String app, TaskLog taskLog) {
        RpcRequest rpcRequest = RpcRequest.builder()
                .interfaces(TaskRpc.class)
                .methodName("sync")
                .parameterTypes(new Class<?>[]{String.class, TaskLog.class})
                .arguments(new Object[]{app, taskLog})
                .build();
        try {
            RpcResponse rpcResponse = clientHolder.removeInvoke(rpcRequest);
            if (rpcResponse != null) {
                throw new RuntimeException(rpcResponse.getException());
            }
            return (Long) rpcResponse.getResult();
        } catch (RpcIOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public boolean cancel(String app, long taskId) {
        RpcRequest rpcRequest = RpcRequest.builder()
                .interfaces(TaskRpc.class)
                .methodName("cancel")
                .parameterTypes(new Class<?>[]{String.class, Long.class})
                .arguments(new Object[]{app, taskId})
                .build();
        try {
            RpcResponse rpcResponse = clientHolder.removeInvoke(rpcRequest);
            if (rpcResponse.getException() != null) {
                throw new RuntimeException(rpcResponse.getException());
            }
            return (Boolean) rpcResponse.getResult();
        } catch (RpcIOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public List<TaskLog> pull(String app, int maxRecord) {
        RpcRequest rpcRequest = RpcRequest.builder()
                .interfaces(TaskRpc.class)
                .methodName("pull")
                .parameterTypes(new Class<?>[]{String.class, Integer.class})
                .arguments(new Object[]{app, maxRecord})
                .build();
        try {
            RpcResponse rpcResponse = clientHolder.removeInvoke(rpcRequest);
            if (rpcResponse.getException() != null) {
                throw new RuntimeException(rpcResponse.getException());
            }
            return (List<TaskLog>) rpcResponse.getResult();
        } catch (RpcIOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void commit(String app, long taskId, boolean success) {
        RpcRequest rpcRequest = RpcRequest.builder()
                .interfaces(TaskRpc.class)
                .methodName("commit")
                .parameterTypes(new Class<?>[]{String.class, Long.class, Boolean.class})
                .arguments(new Object[]{app, taskId, success})
                .build();
        try {
            RpcResponse rpcResponse = clientHolder.removeInvoke(rpcRequest);
            if (rpcResponse.getException() != null) {
                throw new RuntimeException(rpcResponse.getException());
            }
        } catch (RpcIOException e) {
            throw new RuntimeException(e);
        }
    }
}
